package spider.netty.node;

import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spider.netty.client.WebSocketClient;
import spider.netty.client.handler.ClientListener;
import spider.netty.support.RedisManager;
import spider.zookeeper.node.NodeManager;
import spider.zookeeper.util.StringUtils;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 通道节点管理器
 *
 * @author Archx[archx@foxmail.com]
 * @date 2017/4/8.
 */
public class NettyChannelNodeManager {

    private static final Logger logger = LoggerFactory.getLogger(NettyChannelNodeManager.class);

    private ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); // 在线的
    private ChannelGroup clusterChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); // 其他服务器的

    private ConcurrentHashMap<String, ChannelId> uidChannelMap = new ConcurrentHashMap<>(); // 用户通道标识
    private ConcurrentHashMap<ChannelId, String> channelUidMap = new ConcurrentHashMap<>(); // 通道用户标识
    private ConcurrentHashMap<String, ChannelId> clusterNodeChannelMap = new ConcurrentHashMap<>(); // 节点通道标识

    private RedisManager redisManager; // Redis管理
    private NodeManager nodeManager; // 节点管理
    private boolean cluster; // 集群标识
    private String nodeId; // 节点标识

    /**
     * 注册通道
     *
     * @param uid
     * @param channel
     */
    public void register(String uid, Channel channel) {

        channels.add(channel);
        ChannelId channelId = channel.id();

        uidChannelMap.put(uid, channelId);
        channelUidMap.put(channelId, uid);

        if (cluster) {
            redisManager.set(uid, nodeId);
            nodeManager.updateBusy(channels.size());
        }

        logger.info("> Channel is registered: [ channelId = {}, uid = {}]", channelId, uid);
    }

    /**
     * 移除通道
     *
     * @param channel
     */
    public void removeChannel(Channel channel) {
        ChannelId channelId = channel.id();
        String uid = channelUidMap.get(channelId);

        if (!StringUtils.isEmpty(uid)) {
            channelUidMap.remove(channelId);
            uidChannelMap.remove(uid);

            channels.remove(channel);

            if (cluster) {
                redisManager.remove(uid);
                nodeManager.updateBusy(channels.size());
            }

            logger.info("> Channel is disconnected: [ channelId = {}, uid = {}]", channel, uid);
        }
    }

    /**
     * 指定用户发送
     *
     * @param message 消息
     * @param uid     用户标识
     */
    public void writeAndFlush(Object message, String uid) {

        Channel channel = null;

        ChannelId channelId = uidChannelMap.get(uid);
        if (channelId != null) {
            channel = channels.find(channelId);
        }

        // 集群环境
        if (channel == null && cluster) {
            String _nodeId = redisManager.get(uid);
            channel = requireChannel(_nodeId);
        }

        if (channel != null) {
            channel.writeAndFlush(message);
        }
    }

    /**
     * 发送给在线用户
     *
     * @param message
     */
    public void writeAndFlush(Object message) {
        TextWebSocketFrame msg = (TextWebSocketFrame) message;
        channels.writeAndFlush(msg.copy());
        clusterChannels.writeAndFlush(msg.copy());
        ReferenceCountUtil.release(message);
    }

    /**
     * 获取其他服务器通道
     *
     * @param _nodeId 节点标识
     * @return
     */
    private Channel requireChannel(final String _nodeId) {
        if (StringUtils.isEmpty(_nodeId)) {
            return null;
        }

        final ChannelId channelId = clusterNodeChannelMap.get(_nodeId);
        Channel channel = null;
        if (channelId != null) {
            channel = clusterChannels.find(channelId);
        }
        if (channel == null) {
            String url = nodeManager.getNodeUrl(_nodeId);
            WebSocketClient client = new WebSocketClient();
            channel = client.establish(url, new ClientListener() {
                @Override
                public void close() {
                    clusterNodeChannelMap.remove(_nodeId);
                    Channel _channel = clusterChannels.find(channelId);
                    clusterChannels.remove(_channel);
                }
            });

            if (channel != null) {
                clusterChannels.add(channel);
                clusterNodeChannelMap.put(_nodeId, channel.id());
            } else {
                logger.warn("> Can't find server [ nodeId = {}, url = {} ]", _nodeId, url);
            }
        }
        return channel;
    }

    public void setRedisManager(RedisManager redisManager) {
        this.redisManager = redisManager;
    }

    public void setNodeManager(NodeManager nodeManager) {
        this.nodeManager = nodeManager;
    }

    public void setNodeId(String nodeId) {
        this.nodeId = nodeId;
    }

    public void setCluster(boolean cluster) {
        this.cluster = cluster;
    }

    public ChannelGroup getChannels() {
        return channels;
    }

    public ChannelGroup getClusterChannels() {
        return clusterChannels;
    }

    public ConcurrentHashMap<String, ChannelId> getUidChannelMap() {
        return uidChannelMap;
    }

    public ConcurrentHashMap<ChannelId, String> getChannelUidMap() {
        return channelUidMap;
    }

    public ConcurrentHashMap<String, ChannelId> getClusterNodeChannelMap() {
        return clusterNodeChannelMap;
    }

    public RedisManager getRedisManager() {
        return redisManager;
    }

    public NodeManager getNodeManager() {
        return nodeManager;
    }

    public String getNodeId() {
        return nodeId;
    }

    public boolean isCluster() {
        return cluster;
    }
}
